package com.felk.kafka.java;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @Author alison
 * @Date 2024/4/10 21:46
 * @Version 1.0
 * @Description
 */
public class E5_OffsetConsumer2 {

    public static void main(String[] args) {
        System.setProperty("rootLogger.level","info");
        String bootstrapServers = "192.168.56.101:9092";
        String topic = "flink_output_topic";

        // create consumer configs
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "kafkaConsumerGroup");

        // create consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // assign
//        TopicPartition partitionToReadFrom = new TopicPartition(topic, 0);
//        long offsetToReadFrom = 200L;
//        consumer.assign(Arrays.asList(partitionToReadFrom));
        // seek to offset 200
//        consumer.seek(partitionToReadFrom, offsetToReadFrom);

        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        List<TopicPartition> topicPartitionList = partitionInfos.stream()
                .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
                .collect(Collectors.toList());

        consumer.assign(topicPartitionList);
        long offsetToReadFrom = 2L;
        topicPartitionList.forEach(partitionInfo -> consumer.seek(partitionInfo, offsetToReadFrom));

        boolean keepOnReading = true;

        // poll for new data
        while (keepOnReading) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Message received " + record.value() + ", partition " + record.partition() + ", offset=" + record.offset()+", timestamp="+record.timestamp());
            }
        }
    }
}
